// Copyright (c) 2020-present,  INSPUR Co, Ltd.  All rights reserved.
// This source code is licensed under Apache 2.0 License.
#include "db/db_test_util.h"
#include "pure_mem/pmemrep.h"
#include "pure_mem/memoryblock/memory_arena/range_del_mem_arena.h"
#include "pure_mem/rangearena/range_arena.h"
#include "util/random.h"
#include "util/testharness.h"

namespace rocksdb {

    typedef uint64_t Key;

    static const char* Encode(const Key* key) {
      std::string str = std::to_string(*key);
      int len = str.length() + 8 + 1;
      uint32_t keySize = VarintLength(len) + len;
      char* ret = new char[keySize];
      memset(ret , '\0', keySize);
      char*p = EncodeVarint32(ret, len);
      memcpy(p, str.c_str(), str.length());
      return reinterpret_cast<const char*>(ret);
    }

    static Slice Decode(const char* key) {
      Slice buf = GetLengthPrefixedSlice(key);
      Slice mvccKey = Slice(buf.data(), buf.size() - 8);
      const char ts_size = mvccKey[mvccKey.size() - 1];
      if ((size_t)ts_size >= mvccKey.size()) {
        return 0;
      }
      return Slice(buf.data(), mvccKey.size() - 1 - ts_size);
    }



    class TestKeyComparator : public MemTableRep::KeyComparator {
    public:
        typedef rocksdb::Slice DecodedType;

        DecodedType decode_key(const char* key) const override {
          // The format of key is frozen and can be terated as a part of the API
          // contract. Refer to MemTable::Add for details.
          return GetLengthPrefixedSlice(key);
        }

        // Compare a and b. Return a negative value if a is less than b, 0 if they
        // are equal, and a positive value if a is greater than b
        // Compare size in byte order
        int operator()(const char* prefix_len_key1,
                       const char* prefix_len_key2) const override {
          Slice k1 = GetLengthPrefixedSlice(prefix_len_key1);
          Slice k2 = GetLengthPrefixedSlice(prefix_len_key2);

          int comLength = std::min(k1.size(), k2.size());
          for (int i = 0; i < comLength; i++) {
            if ((uint8_t)k1[i] > (uint8_t)k2[i]) {
              return 1;
            } else if ((uint8_t)k1[i] < (uint8_t)k2[i])
              return -1;
          }

          if (k1.size() > k2.size()) return 1;
          if (k1.size() < k2.size()) return -1;
          return 0;
        }

        int operator()(const char* prefix_len_key, const Slice& key) const override {
          Slice key_pre = GetLengthPrefixedSlice(prefix_len_key);
          return key_pre.compare(key);
        }

        ~TestKeyComparator() override = default;
    };

    typedef PureMemRep TestMemArenaRep;
    const size_t standard_key = 1000000;

    class RangeDelMemArenaTest : public testing::Test {
    public:
        static void Insert(TestMemArenaRep* list, Key key) {
          char** buf = nullptr;
          void* rangeArena;
          const auto& keyy = (const Slice&) key;
          const size_t len = sizeof(Key);
          *buf = (char*)list->AllocatePure(len, buf, keyy, &rangeArena);
          memcpy(*buf, &key, sizeof(Key));
          void* buff = *buf;
          list->Insert(buff);
        }

        void Validate(TestMemArenaRep* list) {
          // Check keys exist.
          for (Key key : keys_) {
            ASSERT_TRUE(list->Contains(Encode(&key)));
          }
          // Iterate over the list, make sure keys appears in order and no extra
          // keys exist.
          InlineUserKeyIndex<const MemTableRep::KeyComparator&>::Iterator iter =
              (InlineUserKeyIndex<const MemTableRep::KeyComparator&>::Iterator)
                  reinterpret_cast<const InlineUserKeyIndex<const MemTableRep::KeyComparator &> *>(list->GetIterator());
          ASSERT_FALSE(iter.Valid());
          Key zero = 0;
          iter.Seek(Encode(&zero));
          for (Key key : keys_) {
            ASSERT_TRUE(iter.Valid());
            ASSERT_EQ((char*)key, Decode(iter.key()).data());
            iter.Next();
          }
          ASSERT_FALSE(iter.Valid());
        }

    private:
        std::set<Key> keys_;
    };

    Slice genMVCCKey1(const char* key) {
      uint32_t keySize = strlen(key) + 1;
      char* ret = new char[keySize];
      memset(ret, '\0', keySize);
      memcpy(ret, key, strlen(key));
      return {ret, keySize};
    }

    //测试数据插入
    TEST_F(RangeDelMemArenaTest, MemInsert) {
      const int N = 20000;
      std::set<Key> keys;
      InternalKeyComparator comparator(BytewiseComparator());
      Options options = Options();
      ImmutableCFOptions ioptions(options);
      RangeDelMemArena memoryArena(comparator, ioptions);
      SequenceNumber s = 100000;
      for (int i = 0; i < N; i++) {
        std::string a = std::to_string(s++) + "_test";
        std::string b = std::to_string(s++) + "_west";
        Slice key = Slice(genMVCCKey1(a.data()));
        Slice value = Slice(b.data());

        size_t key_size = key.size();
        size_t internal_key_size = key_size + 8;

        char buf[internal_key_size];
        char* p = buf;
        memcpy(p, key.data(), key_size);
        p += key_size;
        uint64_t packed = PackSequenceAndType(s, kTypeRangeDeletion);
        EncodeFixed64(p, packed);

        Slice slice_key(buf, internal_key_size);

        bool res = memoryArena.Insert(slice_key, value);
        ASSERT_TRUE(res);

        //测试相同key导致插入失败
        if(i % 100 == 0){
          res = memoryArena.Insert(slice_key, value);
          ASSERT_FALSE(res);
        }
      }
    }

    //测试数据插入，插入后拿出的key,value是否和插入前相同（测试Seek）
    TEST_F(RangeDelMemArenaTest, MemInsertAndSeek) {
      const int N = 20000;
      std::set<Key> keys;
      Options options = Options();;
      ImmutableCFOptions ioptions(options);
      InternalKeyComparator comparator(BytewiseComparator());
      RangeDelMemArena memoryArena(comparator, ioptions);
      SequenceNumber s = 100000;
      for (int i = 0; i < N; i++) {
        std::string a = std::to_string(s++) + "_test";
        std::string b = std::to_string(s++) + "_west";
        Slice key = Slice(genMVCCKey1(a.data()));
        Slice value = Slice(b.data());

        size_t key_size = key.size();
        size_t internal_key_size = key_size + 8;

        char buf[internal_key_size];
        char* p = buf;
        memcpy(p, key.data(), key_size);
        p += key_size;
        uint64_t packed = PackSequenceAndType(s, kTypeRangeDeletion);
        EncodeFixed64(p, packed);

        Slice slice_key(buf, internal_key_size);


        bool insert_res = memoryArena.Insert(slice_key, value);
        ASSERT_TRUE(insert_res);
        auto mem_iter = memoryArena.Seek(slice_key);     //测试内存存储区的Seek
        if(mem_iter->Valid()){
          Slice mem_key = GetLengthPrefixedSlice(mem_iter->key());
          Slice mem_value = GetLengthPrefixedSlice(mem_key.data() + mem_key.size());   //Seek单独for循环
          int res = mem_key.compare(slice_key);
          int res1 = mem_value.compare(value);
          if(res != 0){
            std::cout<<"mem插入前后userkey不同， 插入mem:"<<slice_key.ToString()<<"查到："<<mem_key.ToString()<<std::endl;
            exit(1);
          }
          if(res1 != 0){
            std::cout<<"mem插入前后value不同， 插入mem:"<<value.ToString()<<"查到："<<mem_value.ToString()<<std::endl;
            exit(1);
          }
        }
      }

      s = 100000;
      //Seek单独for循环
      for (int i = 0; i < N; i++) {
        std::string a = std::to_string(s++) + "_test";
        std::string b = std::to_string(s++) + "_west";
        Slice key = Slice(genMVCCKey1(a.data()));
        Slice value = Slice(b.data());

        size_t key_size = key.size();
        size_t internal_key_size = key_size + 8;

        char buf[internal_key_size];
        char* p = buf;
        memcpy(p, key.data(), key_size);
        p += key_size;
        uint64_t packed = PackSequenceAndType(s, kTypeRangeDeletion);
        EncodeFixed64(p, packed);
        Slice slice_key(buf, internal_key_size);


        auto mem_iter = memoryArena.Seek(slice_key);     //测试内存存储区的Seek
        if(mem_iter->Valid()){
          Slice mem_key = GetLengthPrefixedSlice(mem_iter->key());
          Slice mem_value = GetLengthPrefixedSlice(mem_key.data() + mem_key.size());
          int res = mem_key.compare(slice_key);
          int res1 = mem_value.compare(value);
          if(res != 0){
            std::cout<<"mem插入前后userkey不同， 插入mem:"<<slice_key.ToString()<<"查到："<<mem_key.ToString()<<std::endl;
            exit(1);
          }
          if(res1 != 0){
            std::cout<<"mem插入前后value不同， 插入mem:"<<value.ToString()<<"查到："<<mem_value.ToString()<<std::endl;
            exit(1);
          }
        }
      }

    }


}  //namespace rocksdb

int main(int argc, char** argv) {
  ::testing::InitGoogleTest(&argc, argv);
  return RUN_ALL_TESTS();
}